package com.lanccj.aep.nbiot.service;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.ctiot.aep.mqmsgpush.sdk.IMsgConsumer;
import com.ctiot.aep.mqmsgpush.sdk.IMsgListener;
import com.ctiot.aep.mqmsgpush.sdk.MqMsgConsumer;
import com.lanccj.aep.nbiot.callback.AepMqDecodeCallback;
import com.lanccj.aep.nbiot.callback.AepMqOrginCallback;
import com.lanccj.aep.nbiot.constant.AepConstant;
import com.lanccj.aep.nbiot.mq.*;
import com.lanccj.aep.nbiot.propertie.AepIotProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * @author chenjian_yq(lanccj)
 * @version 1.0
 * @description: AER MQ服务
 * @date 2021/11/18 8:26
 */
public class AepMqService {

    Logger log = LoggerFactory.getLogger(AepMqService.class);

    /**
     * AepIot配置对象
     */
    AepIotProperties aepIotProperties;

    public AepMqService(AepIotProperties aepIotProperties){
        this.aepIotProperties = aepIotProperties;
    }

    /**
     * 监听AEP MQ消息(该配置下所有的topic信息) 某个租户的所有消息
     * @param topicNames 需要监听的topicName信息，所有直接不配置空即可
     * @param callback 监听后回调函数
     */
    public void listenerOrign(List<String> topicNames, AepMqOrginCallback callback){
        String server = aepIotProperties.getMq().getServer()==null?"msgpush.ctwing.cn:16651":aepIotProperties.getMq().getServer(); //消息服务地址
        String tenantId = aepIotProperties.getMq().getTenantId();//租户ID
        String token = aepIotProperties.getMq().getToken();//身份认证token串
        String certFilePath = ""; //直接填空字符串，CA证书，JDK已经内置相关根证书，无需指定

        //创建消息接收Listener
        IMsgListener msgListener = new IMsgListener() {

            @Override
            public void onMessage(String msg) {
                //接收消息
                System.out.println(msg);
                //消息处理...
                //为了提高效率，建议对消息进行异步处理（使用其它线程、发送到Kafka等）
                callback.listener(msg);
            }
        };

        //创建消息接收类
        IMsgConsumer consumer = new MqMsgConsumer();
        try {
            //初始化
            /**
             * @param server  消息服务server地址
             * @param tenantId 租户Id
             * @param token    用户认证token
             * @param certFilePath 证书文件路径
             * @param topicNames   主题名列表，如果该列表为空或null，则自动消费该租户所有主题消息
             * @param msgListener 消息接收者
             * @return 是否初始化成功
             */
            consumer.init(server, tenantId, token, certFilePath, topicNames, msgListener);

            //开始接收消息
            consumer.start();

            //程序退出时，停止接收、销毁
            //consumer.stop();
            //consumer.destroy();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public void listenerDecode(List<String> topicNames, AepMqDecodeCallback callback){
        String server = aepIotProperties.getMq().getServer()==null?"msgpush.ctwing.cn:16651":aepIotProperties.getMq().getServer(); //消息服务地址
        String tenantId = aepIotProperties.getMq().getTenantId();//租户ID
        String token = aepIotProperties.getMq().getToken();//身份认证token串
        String certFilePath = ""; //直接填空字符串，CA证书，JDK已经内置相关根证书，无需指定

        //创建消息接收Listener
        IMsgListener msgListener = new IMsgListener() {

            @Override
            public void onMessage(String msg) {
                //接收消息
                System.out.println(msg);
                //消息处理...
                //为了提高效率，建议对消息进行异步处理（使用其它线程、发送到Kafka等）
                //先判断是什么消息
                JSONObject jsonObject = JSONUtil.parseObj(msg);
                String messageType = (String) jsonObject.get(AepConstant.MqMsgDefineKey.MessageType);
                if (messageType.equals(AepConstant.MqMsgType.DataReport)) {
                    log.info("消息:{}开始转换为:{}对象",msg,"MqDataReport");
                    callback.listener(JSON.parseObject(msg, MqDataReport.class));
                }else if (messageType.equals(AepConstant.MqMsgType.OnlineOffReport)) {
                    log.info("消息:{}开始转换为:{}对象",msg,"MqOnlineOffReport");
                    callback.listener(JSON.parseObject(msg, MqOnlineOffReport.class));
                }else if (messageType.equals(AepConstant.MqMsgType.CommandResponse)) {
                    log.info("消息:{}开始转换为:{}对象",msg,"MqCommandResponse");
                    callback.listener(JSON.parseObject(msg, MqCommandResponse.class));
                }else if (messageType.equals(AepConstant.MqMsgType.AddDeleteDevice)) {
                    log.info("消息:{}开始转换为:{}对象",msg,"AddDeleteDevice");
                    callback.listener(JSON.parseObject(msg, MqAddDeleteDeviceReport.class));
                }else if (messageType.equals(AepConstant.MqMsgType.EventType)) {
                    log.info("消息:{}开始转换为:{}对象",msg,"EventType");
                    MqEventReport mqEventReport = JSON.parseObject(msg, MqEventReport.class);
                    callback.listener(mqEventReport);
                }else{
                    try {
                        throw new Exception(messageType+"该消息类型暂未接入，请联系开发拓展下吧!");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }

            }
        };

        //创建消息接收类
        IMsgConsumer consumer = new MqMsgConsumer();
        try {
            //初始化
            /**
             * @param server  消息服务server地址
             * @param tenantId 租户Id
             * @param token    用户认证token
             * @param certFilePath 证书文件路径
             * @param topicNames   主题名列表，如果该列表为空或null，则自动消费该租户所有主题消息
             * @param msgListener 消息接收者
             * @return 是否初始化成功
             */
            consumer.init(server, tenantId, token, certFilePath, topicNames, msgListener);

            //开始接收消息
            consumer.start();


        } catch (Exception e) {
            e.printStackTrace();
            //程序退出时，停止接收、销毁
            //consumer.stop();
            //consumer.destroy();
        }
    }

}
